package com.miao.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * @Author <a href="https://gitee.com/a-tom-is-cry">Xing</a>
 * @Date 2024/11/7 17:15
 * @Description
 */
@Component
public class Consumer implements ConsumerSeekAware {
    @KafkaListener(topics = {"testTopic"})
    public void onMessageReceived(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.topic() + "====>" + record.value());
        }

        ack.acknowledge();
    }

//    @Override
//    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
//        //按照时间戳设置偏移
//        callback.seekToTimestamp(assignments.keySet(),1670233826705L);
//        //设置偏移到最近
//        callback.seekToEnd(assignments.keySet());
//        //设置偏移到最开始
//        callback.seekToBeginning(assignments.keySet());
//        //指定 offset
//        for (TopicPartition topicPartition : assignments.keySet()) {
//            callback.seek(topicPartition.topic(),topicPartition.partition(),0L);
//        }
//    }
}
